package com.edu.yx.app.App_04_DwsApp.Dws_01_Traffic;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp_moreTopic;
import com.edu.yx.bean.TrafficPageViewBean;
import com.edu.yx.bean.TrafficPageViewResultBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashMap;

import static com.edu.yx.common.Constant.*;

public class Dws_01_Traffic_Chanel_Page_View_Window extends BaseApp_moreTopic {


    public static void main(String[] args) {
        new Dws_01_Traffic_Chanel_Page_View_Window().init(
                4001,
                2,
                "Dws_01_Traffic_Chanel_Page_View_Window",

                //todo 继承多主题的BaseApp_moreTopic
                TOPIC_DWD_TRAFFIC_PAGE,
                TOPIC_DWD_TRAFFIC_UV_DETAIL,
                TOPIC_DWD_TRAFFIC_UJ_DETAIL
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {

        //1.由于三个主题的内容都通过同一个pojo进行封装，所以结果用union合并
        DataStream<TrafficPageViewBean> unionAllStream = unionAll(streams);

        //2.开窗聚合 todo (问题比较大)
        SingleOutputStreamOperator<TrafficPageViewBean> windowAggStream = WindowAgg(unionAllStream);

        //3.将聚合结果进一步计算，并且封装
        SingleOutputStreamOperator<TrafficPageViewResultBean> parseToPojoStream = parseToPojo(windowAggStream);

        //4.写入doris
       writeDoris(parseToPojoStream);

    }

    private void writeDoris(SingleOutputStreamOperator<TrafficPageViewResultBean> resultStream) {
        resultStream
                .map(bean ->{
                    //配置json可以将驼峰转换成带下划线，然后方便导入数据到doris
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);
//                    System.out.println(json);
                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_traffic_ch_page_view_window"));

    }

    private SingleOutputStreamOperator<TrafficPageViewResultBean> parseToPojo(SingleOutputStreamOperator<TrafficPageViewBean> restultStream) {

        SingleOutputStreamOperator<TrafficPageViewResultBean> streamOperator = restultStream
                .filter(bean -> bean.getSvCt() != 0L)
                .map(bean -> {

                    //分母
                    Long svCt = bean.getSvCt();

                    //分子
                    Long pvCt = bean.getPvCt();
                    Long durSum = bean.getDurSum();
                    Long ujCt = bean.getUjCt();


                    Double avgPv = Double.valueOf((pvCt * 100) / svCt) / 100.0;
                    Double avgDurSum = Double.valueOf((durSum * 100) / svCt) / 100.0;
                    Double ujRate = Double.valueOf((ujCt * 100) / svCt) / 100.0;


                    TrafficPageViewResultBean trafficPageViewResultBean = new TrafficPageViewResultBean(
                            bean.getStt(),
                            bean.getEdt(),
                            bean.getCh(),
                            bean.getCurDate(),
                            bean.getUvCt(),
                            bean.getSvCt(),
                            avgPv,
                            avgDurSum,
                            ujRate,
                            bean.getTs()
                    );


                    System.out.println(trafficPageViewResultBean + "  " + bean.getUjCt());

                    return trafficPageViewResultBean;
                });

        return streamOperator;


    }

    private SingleOutputStreamOperator<TrafficPageViewBean> WindowAgg(DataStream<TrafficPageViewBean> unionAllStream) {

        //由于已结union到一起了，直接上水位线
        SingleOutputStreamOperator<TrafficPageViewBean> result = unionAllStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //乱序容忍越大，时效性越差
                                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                                .withTimestampAssigner((tpb, ts) -> tpb.getTs())
                )
                // 分组：渠道粒度
                .keyBy(union ->  union.getCh() )
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //捕获迟到数据
                .sideOutputLateData(new OutputTag<TrafficPageViewBean>("late") {
                })
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean value1,
                                                              TrafficPageViewBean value2) throws Exception {

                                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                                value1.setUjCt(value1.getUjCt() + value2.getUjCt());

                                return value1;
                            }
                            //获取sst、edt、curdate
                        },
                        new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                            @Override
                            public void process(String keyBy, //前面分完组的key
                                                Context ctx,
                                                Iterable<TrafficPageViewBean> elements,
                                                Collector<TrafficPageViewBean> out) throws Exception {

                                //获取前面聚合计算结果的数据
                                TrafficPageViewBean bean = elements.iterator().next();

                                bean.setStt(TempUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));

                                if (bean.getSvCt()!=0){
                                    out.collect(bean);
                                }


                            }
                        }
                );

        /**
         * 分析迟到原因：
         * 由于uv、uj的先经历了一次计算，会有一定延迟，后面汇总又进行一次窗口计算
         * 而直接从dwd_page拿的数据只进行一次计算汇总
         * uv、uj的数据传输上有物理时间上的延迟，可能会被dwd_page直接拿到的数据先关闭窗口
         */
//       result.print("main");
//       result.getSideOutput(new OutputTag<TrafficPageViewBean>("late"){}).print("late");

       return result;
    }

    private DataStream<TrafficPageViewBean> unionAll(HashMap<String, DataStreamSource<String>> streams) {

        //1.TOPIC_DWD_TRAFFIC_PAGE todo 记录首次会话数，记录页面，记录留存时间
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPage_Stream = streams
                //todo 这个流负责读取会话数，页面浏览数，累积访问时间
                .get(TOPIC_DWD_TRAFFIC_PAGE)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    //todo 从大的json对象内获取小的对象
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    //todo 读取common中的数据，按照需求获取需要的数据
                    String ch = common.getString("ch");

                    //todo 用于下游统计的数据
                    Long uvCt = 0L;
                    //一个会话，如果last_page_id为空，说明还是第一个会话，拿到，否则为0
                    //todo 一个会话，如果last_page_id不存在，那么他的长度为0，说明也是第一个会话，拿到，否则为0
                    Long svCt = page.getString("last_page_id") == null || page.getString("last_page_id").length()==0 ? 1L : 0L;
                    Long pvCt = 1L; //进来一个页面就计算一次1
                    Long durSum = page.getLong("during_time");
                    Long ujCt = 0L;

                    //todo 用于提供进行窗口聚合的事件时间
                    Long ts = obj.getLong("ts");

                    return new TrafficPageViewBean(
                            "", //下游窗口计算生成
                            "",
                            ch,
                            "", //下游计算生成
                            uvCt,
                            svCt,
                            pvCt,
                            durSum,
                            ujCt,
                            ts

                    );
                });

        //2.TOPIC_DWD_TRAFFIC_UV_DETAIL todo 读取uv
        SingleOutputStreamOperator<TrafficPageViewBean> trafficUV_Stream = streams
                //todo 这个流负责读取会话数，页面浏览数，累积访问时间
                .get(TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    //todo 从大的json对象内获取小的对象
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    //todo 读取common中的数据，按照需求获取需要的数据

                    String ch = common.getString("ch");

                    //todo 用于下游统计的数据
                    Long uvCt = 1L;
                    //一个会话，如果last_page_id为空，说明还是第一个会话，拿到，否则为0
                    Long svCt = 0L;
                    Long pvCt = 0L; //进来一个页面就计算一次1
                    Long durSum = 0L;
                    Long ujCt = 0L;


                    //todo 用于提供进行窗口聚合的事件时间
                    Long ts = obj.getLong("ts");

                    return new TrafficPageViewBean(
                            "", //下游窗口计算生成
                            "",
                            ch,
                            "", //下游计算生成
                            uvCt,
                            svCt,
                            pvCt,
                            durSum,
                            ujCt,
                            ts
                    );
                });

        //3.TOPIC_DWD_TRAFFIC_UJ_DETAIL todo 读取uj
        SingleOutputStreamOperator<TrafficPageViewBean> trafficUJ_Stream = streams
                //todo 这个流负责读取会话数，页面浏览数，累积访问时间
                .get(TOPIC_DWD_TRAFFIC_UJ_DETAIL)
                .map(strjson -> {

                    //todo 将kafka拿到的json格式的字符串转换成json对象
                    JSONObject obj = JSONObject.parseObject(strjson);

                    //todo 从大的json对象内获取小的对象
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    //todo 读取common中的数据，按照需求获取需要的数据
                    String ch = common.getString("ch");

                    //todo 用于下游统计的数据
                    Long uvCt = 0L;
                    //一个会话，如果last_page_id为空，说明还是第一个会话，拿到，否则为0
                    Long svCt = 0L;
                    Long pvCt = 0L; //进来一个页面就计算一次1
                    Long durSum = 0L;
                    Long ujCt = 1L;
                    Double ujRate = 0.0;

                    //todo 用于提供进行窗口聚合的事件时间
                    Long ts = obj.getLong("ts");

                    return new TrafficPageViewBean(
                            "", //下游窗口计算生成
                            "",
                            ch,
                            "", //下游计算生成
                            uvCt,
                            svCt,
                            pvCt,
                            durSum,
                            ujCt,
                            ts
                    );
                });

        //4.union一下
        return trafficPage_Stream.union(trafficUV_Stream, trafficUJ_Stream);

    }
}
